package com.derbysoft.lowcostadapter.commons.redis.service.impl;

import com.derbysoft.lowcostadapter.commons.redis.service.JedisClusterService;
import com.derbysoft.lowcostadapter.commons.redis.service.RedisSerializable;
import com.derbysoft.lowcostadapter.commons.redis.utils.RedisSerializableUtils;
import com.derbysoft.redis.clients.normal.SingleJedis;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;
import redis.clients.jedis.PipelineBlock;
import redis.clients.util.SafeEncoder;

import java.util.*;

/**
 * Created by: jason
 * Date: 2012-07-05
 */
public class JedisClusterServiceImpl implements JedisClusterService, InitializingBean {

    private static Log logger = LogFactory.getLog(JedisClusterServiceImpl.class);

    private String redisHosts;

    private List<SingleJedis> jedises = new ArrayList<SingleJedis>();

    @Override
    public void saveHSetEntities(final List<HSetEntity> hSetEntities) {
        try {
            jedises.get(0).pipelined(new PipelineBlock() {
                @Override
                public void execute() {
                    for (HSetEntity hSetEntity : hSetEntities) {
                        hset(hSetEntity.getKey(), hSetEntity.getField(), hSetEntity.getValue());
                        if (hSetEntity.getExpireTime() != null) {
                            expireAt(hSetEntity.getKey(), hSetEntity.getExpireTime());
                        }
                    }
                }
            });
        } catch (Exception ex) {
            logger.error(ex.getMessage(), ex);
        }
    }

    @Override
    public void set(RedisSerializable object) {
        try {
            jedises.get(0).set(object.toKey(), object.value());
        } catch (Exception ex) {
            logger.error(ex.getMessage(), ex);
        }
    }

    @Override
    public void removeAll(String key) {
        try {
            jedises.get(0).expire(key, 0);
        } catch (Exception ex) {
            logger.error(ex.getMessage(), ex);
        }
    }

    @Override
    public void remove(String key, String field) {
        try {
            jedises.get(0).hdel(key, new String[]{field});
        } catch (Exception ex) {
            logger.error(ex.getMessage(), ex);
        }
    }

    @Override
    public boolean hExist(String key, String field) {
        try {
            return jedises.get(0).hexists(key, field);
        } catch (Exception ex) {
            logger.error(ex.getMessage(), ex);
        }
        return false;
    }

    @Override
    public void flushAll() {
        try {
            jedises.get(0).flushAll();
        } catch (Exception ex) {
            logger.error(ex.getMessage(), ex);
        }
    }

    @Override
    public <E extends RedisSerializable> void pipelined(final List<E> objects) {
        try {
            jedises.get(0).pipelined(new PipelineBlock() {
                @Override
                public void execute() {
                    for (E object : objects) {
                        hset(object.toKey(), object.toFiled(), object.value());
                        if (object.expireTime() != null) {
                            expireAt(object.toKey(), object.expireTime());
                        }
                    }
                }
            });
        } catch (Exception ex) {
            logger.error(ex.getMessage(), ex);
        }
    }

    @Override
    public void hset(RedisSerializable object) {
        try {
            jedises.get(0).hset(object.toKey(), object.toFiled(), object.value());
            if (object.expireTime() != null) {
                jedises.get(0).expireAt(object.toKey(), object.expireTime());
            }
        } catch (Exception ex) {
            logger.error(ex.getMessage(), ex);
        }
    }

    @Override
    public String get(String key) {
        try {
            return getSingleJedis().get(key);
        } catch (Exception ex) {
            logger.error(ex.getMessage(), ex);
        }
        return null;
    }

    @Override
    public Map<String, RedisSerializable> hGetAll(String key) {
        TreeMap<String, RedisSerializable> objectMap = new TreeMap<String, RedisSerializable>();
        try {
            Map<String, String> values = getSingleJedis().hgetAll(key);
            for (Map.Entry<String, String> entries : values.entrySet()) {
                if (StringUtils.isBlank(entries.getValue())) {
                    continue;
                }
                objectMap.put(entries.getKey(), RedisSerializableUtils.of(key, entries.getKey(), entries.getValue()));
            }
        } catch (Exception ex) {
            logger.error(ex.getMessage(), ex);
        }
        return objectMap;
    }

    @Override
    public String hGet(String key, String field) {
        try {
            return getSingleJedis().hget(key, field);
        } catch (Exception ex) {
            logger.error(ex.getMessage(), ex);
        }
        return null;
    }

    @Override
    public Collection<String> keys(String pattern) {
        Set<String> keys = new HashSet<String>();
        try {
            for (String key : getSingleJedis().keys(pattern)) {
                keys.add(SafeEncoder.encode(SafeEncoder.encode(key)));
            }
            return keys;
        } catch (Exception ex) {
            logger.error(ex.getMessage(), ex);
        }
        return keys;
    }

    @Override
    public void afterPropertiesSet() {
        if (StringUtils.isBlank(redisHosts)) {
            throw new IllegalArgumentException("Cluster's redisHost is blank.");
        }
        for (String host : redisHosts.split(",")) {
            jedises.add(new SingleJedis(host));
        }
        if (CollectionUtils.isEmpty(jedises)) {
            throw new IllegalArgumentException("Cluster's redis is empty.");
        }
    }

    @Override
    public void saveSetEntities(final List<SetEntity> setEntities) {
        try {
            jedises.get(0).pipelined(new PipelineBlock() {
                @Override
                public void execute() {
                    for (SetEntity setEntity : setEntities) {
                        set(setEntity.getKey(), setEntity.getValue());
                    }
                }
            });
        } catch (Exception ex) {
            logger.error(ex.getMessage(), ex);
        }
    }

    private SingleJedis getSingleJedis() {
        if (jedises.size() == 1) {
            return jedises.get(0);
        }
        return jedises.get((int) Math.ceil(Math.random() * jedises.size()) - 1);
    }

    public void setRedisHosts(String redisHosts) {
        this.redisHosts = redisHosts;
    }
}
